package com.lnj.community.event;

import com.lnj.community.dao.entity.DiscussPost;
import com.lnj.community.dao.entity.Event;
import com.lnj.community.dao.entity.Message;
import com.lnj.community.service.DiscussPostService;
import com.lnj.community.service.ElasticsearchService;
import com.lnj.community.service.MessageService;
import com.lnj.community.utiles.CommunityConstant;
import com.lnj.community.utiles.OSSUtile;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.stereotype.Component;
import com.alibaba.fastjson.JSONObject;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledFuture;

/**
 * 消费者
 */
@Component
public class EventConsumer implements CommunityConstant {
    private static final Logger logger= LoggerFactory.getLogger(EventConsumer.class);
    @Autowired
    private MessageService messageService;
    @Autowired
    private DiscussPostService discussPostService;
    @Autowired
    private ElasticsearchService elasticsearchService;
    @Value("${wk.image.storage}")
    private String wkImageStorage;
    @Value("${wk.image.command}")
    private String wkImageCommand;
    @Autowired
    private ThreadPoolTaskScheduler taskScheduler;

    /**
     * 处理事件  监听指定的主题 点赞 评论 关注
     * @param record
     */
    // 消费评论、点赞、关注事件
    @KafkaListener(topics = {TOPIC_COMMENT,TOPIC_FOLLOW,TOPIC_LIKE})
    public void handleEvent(ConsumerRecord  record){
        if (record == null || record.value() == null) {
            logger.error("消息的内容为空!");
            return;
        }

        Event event = JSONObject.parseObject(record.value().toString(), Event.class);
        if (event == null) {
            logger.error("消息格式错误!");
            return;
        }

        // 发送站内通知 就是用户把消息发送给系统，然后系统统一安排发送给用户
        Message message = new Message();
        message.setFromId(SYSTEM_USER_ID);
        message.setToId(event.getEntityUserId());
        message.setConversationId(event.getTopic());
        message.setCreateTime(new Date());

        Map<String, Object> content = new HashMap<>();
        content.put("userId", event.getUserId());
        content.put("entityType", event.getEntityType());
        content.put("entityId", event.getEntityId());

        if (!event.getData().isEmpty()) {
            for (Map.Entry<String, Object> entry : event.getData().entrySet()) {
                content.put(entry.getKey(), entry.getValue());
            }
        }

        message.setContent(JSONObject.toJSONString(content));
        messageService.addMessage(message);
    }

    /**
     * 处理事件  监听指定的主题 elasticsearch的数据修改增删改查  异步处理  把数据放到Elasticsearch中
     * 处理事件  监听指定的主题  发布帖子  异步处理  把数据放到Elasticsearch中
     * @param record
     */
    // 消费发帖事件
    @KafkaListener(topics = {TOPIC_PUBLISH})
    public void handlePublishMessage(ConsumerRecord record) {
        if (record == null || record.value() == null) {
            logger.error("消息的内容为空!");
            return;
        }

        Event event = JSONObject.parseObject(record.value().toString(), Event.class);
        if (event == null) {
            logger.error("消息格式错误!");
            return;
        }
        // 首先要把帖子查出来 然后放到Elasticsearch中
        DiscussPost post = discussPostService.findDiscussPostById(event.getEntityId());
        elasticsearchService.saveDiscussPost(post);
    }

    // 消费删帖事件
    @KafkaListener(topics = {TOPIC_DELETE})
    public void handleDeleteMessage(ConsumerRecord record) {
        if (record == null || record.value() == null) {
            logger.error("消息的内容为空!");
            return;
        }
        Event event = JSONObject.parseObject(record.value().toString(), Event.class);
        if (event == null) {
            logger.error("消息格式错误!");
            return;
        }
        // 把帖子从Elasticsearch中删除
        elasticsearchService.deleteDiscussPost(event.getEntityId());
    }

    @KafkaListener(topics = {TOPIC_SHARE})
    public void handleShareMessage(ConsumerRecord record) {
        if (record == null || record.value() == null) {
            logger.error("消息的内容为空!");
            return;
        }
        Event event = JSONObject.parseObject(record.value().toString(), Event.class);
        if (event == null) {
            logger.error("消息格式错误!");
            return;
        }
        // 生成分享图片的名称
        String htmlUrl = (String) event.getData().get("htmlUrl");
        String fileName = (String) event.getData().get("fileName");
        String suffix = (String) event.getData().get("suffix");

        String cmd=wkImageCommand+" --quality 75 "+htmlUrl+" "+wkImageStorage+"/"+fileName+suffix;
        try {
            Runtime.getRuntime().exec(cmd);
            logger.info("分享图片生成成功:"+cmd);
        }catch (Exception e) {
            logger.error("分享图片生成失败:"+e.getMessage());
            e.printStackTrace();
        }
        //启动定时器 把监视的图片生成成功就立马传到阿里云上面
        UpdateTask task=new UpdateTask(fileName,suffix);
        //启动定时器 500毫秒后执行任务 每500毫秒执行一次任务
        ScheduledFuture<?> scheduledFuture = taskScheduler.scheduleAtFixedRate(task, 500);
        //如果定时器
        task.setFuture(scheduledFuture);
    }

    class UpdateTask implements Runnable {
        String fileName;
        String suffix;
        //启动任务的返回值  可以取消任务
        Future future;
        //上传时间
        private Long uploadTime;
        //上传次数
        private int uploadCount;
        public UpdateTask(String fileName,String suffix) {
            this.fileName=fileName;
            this.suffix=suffix;
            this.uploadTime=System.currentTimeMillis();
            this.uploadCount=0;
        }
        public void setFuture(Future future) {
            this.future=future;
        }
        @Override
        public void run() {
            //生成图片失败
            if (System.currentTimeMillis()-uploadTime>30000) {
                logger.error("执行时间过长，终止任务:"+fileName);
                future.cancel(true);
                return;
            }
            //上传次数过多
            if (uploadCount>=3) {
                logger.error("上传次数过多，终止任务:"+fileName);
                future.cancel(true);
                return;
            }
            //上传图片
            String path=wkImageStorage+"/"+fileName+suffix;
            File file=new File(path);
            if (file.exists()) {
                uploadCount++;
                logger.info("第"+uploadCount+"次开始上传图片:"+fileName);
                try {
                    String url= OSSUtile.uploadToAliyun(new FileInputStream(file), fileName+suffix);
                    if (url!=null) {
                        logger.info("第"+uploadCount+"次上传图片成功:"+fileName);
                        future.cancel(true);
                    }
                } catch (FileNotFoundException e) {
                    logger.info("第"+uploadCount+"次开始上传图片失败:"+fileName);
                }
            }else{
                logger.info("第"+uploadCount+"次开始上传图片失败:"+fileName);
            }
        }
    }

}
